package cn.ipanel.bigdata.boot.source.genre

import cn.ipanel.bigdata.boot.date.Time
import cn.ipanel.bigdata.boot.logger.Logger
import cn.ipanel.bigdata.boot.source.{DataSource, LocalExec, Mapper, SparkExec, Table}
import cn.ipanel.bigdata.utils.Util
import org.apache.commons.lang3.exception.ExceptionUtils
import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode}

/**
 * Author: lzz
 * Date: 2021/11/16 16:15
 */
abstract class Mysql(dbName: String, tbName: String) extends Mapper with Table with SparkExec with LocalExec {

  import Mysql._

  private[this]  lazy val SC: SQLContext = spark.sqlContext
  // scala 低版本可以这么用
//  private[this]  lazy val SC: SQLContext = SQLContext.getOrCreate(spark.sparkContext)

  override def getSourceGenre: DataSource.Genre = DataSource.GENRE_MYSQL
  override def getSourceModel: DataSource.Model = DataSource.MODEL_WRITE
  override def getDBName: String = dbName
  override def getTBName: String = tbName

  override def _local_get(sql: String): Seq[Map[String, String]] = {
    Logger.I(s"Local_Get Mysql[$getDBName.$getTBName], Sql => $sql")
    Seq()
  }

  override def _local_sql(sql: String): Int = {
    Logger.I(s"Local_Sql Mysql[$getDBName.$getTBName], Sql => $sql")
    0
  }

  override def exec(sql: String): DataFrame = {
    Logger.I(s"Exec Mysql[$getDBName.$getTBName], Sql = $sql")
    val time = Time.now
    try {
      SC.read
        .format(protocol.source)
        .options(Map(PARAM_URL -> database.url,
                     PARAM_DRIVER -> protocol.driver,
                     PARAM_USER -> database.username,
                     PARAM_PASSWORD -> database.password,
                     PARAM_TABLE -> sql))
        .load()
    } catch {
      case e: Exception =>
        Logger.E(
          s"""Exec Mysql[$getDBName.$getTBName] Failed.
             | Because: ${ExceptionUtils.getStackTrace(e)}
             |""".stripMargin)
        emptyTable.toDF()
    }
  }

  override def load: DataFrame = {
    Logger.I(s"Load Mysql[$getDBName.$getTBName]")
    try {
      SC.read
        .format(protocol.source)
        .options(Map(PARAM_URL -> database.url,
                     PARAM_DRIVER -> protocol.driver,
                     PARAM_USER -> database.username,
                     PARAM_PASSWORD -> database.password,
                     PARAM_TABLE -> getTBName))
        .load()
    } catch {
      case e: Exception =>
        Logger.E(
          s"""Load Mysql[$getDBName.$getTBName] Failed.
             | Because: ${ExceptionUtils.getStackTrace(e)}
             |""".stripMargin)
        null
    }
  }

  override def save(df: DataFrame, mode: SaveMode = SaveMode.Overwrite): Unit = {
    if(df != null){
      getSourceModel match {
        case DataSource.MODEL_WRITE =>
          Logger.I(s"Save Mysql[$getDBName.$getTBName]")
          try {
            df.write
              .format(protocol.source)
              .mode(mode)
              .options(Map(PARAM_URL -> database.url,
                           PARAM_DRIVER -> protocol.driver,
                           PARAM_USER -> database.username,
                           PARAM_PASSWORD -> database.password,
                           PARAM_TABLE -> getTBName))
              .save()
          } catch {
            case e: Exception =>
              Logger.E(
                s"""Save Mysql[$getDBName.$getTBName] Failed.
                   | Because: ${ExceptionUtils.getStackTrace(e)}
                   |""".stripMargin)
          }
        case _ =>
          Logger.I(s"Refuse Save Mysql[$getDBName.$getTBName]. Because: Only Read")
      }
    }
  }
}

protected[source] object Mysql {

  val PARAM_URL       : String = "url"
  val PARAM_DRIVER    : String = "driver"
  val PARAM_USER      : String = "user"
  val PARAM_PASSWORD  : String = "password"
  val PARAM_TABLE     : String = "dbtable"
}